package mysql

import (
	"database/sql"
	"fmt"
	"strings"

	"github.com/example-user/go-pocket-etl/pkg/core"
	"github.com/example-user/go-pocket-etl/pkg/factory"
	// 匿名导入 MySQL 驱动，使其通过 init() 函数将自己注册到 database/sql 包中。
	_ "github.com/go-sql-driver/mysql"
)

// init 实现了组件的自注册。
func init() {
	factory.RegisterSink("mysql", NewSink)
}

// Sink 实现了 core.Sink 接口，用于将数据批量写入到 MySQL 数据库。
// 它通过在一个事务中构建并执行一个大的多行 INSERT 语句来实现高效的批量写入。
type Sink struct {
	db            *sql.DB           // 数据库连接池。
	table         string            // 目标表名。
	columnMapping map[string]string // 列映射关系 (Record 中的键 -> 数据库中的列名)。
}

// NewSink 是 mysql.Sink 的构造函数，由工厂调用。
func NewSink() core.Sink {
	return &Sink{}
}

// Open 负责解析配置并建立数据库连接。
//
// YAML 配置示例:
//
//	sink:
//	  type: mysql
//	  params:
//	    dsn: "user:password@tcp(127.0.0.1:3306)/database"
//	    table: "users"
//	    column_mapping:
//	      "id": "user_id"
//	      "email": "user_email"
//	      "status": "account_status"
func (s *Sink) Open(config map[string]interface{}) error {
	// 'dsn' 是必需配置。
	dsn, ok := config["dsn"].(string)
	if !ok || dsn == "" {
		return fmt.Errorf("mysql sink: config is missing or has invalid 'dsn'")
	}

	// 'table' 是必需配置。
	table, ok := config["table"].(string)
	if !ok || table == "" {
		return fmt.Errorf("mysql sink: config is missing or has invalid 'table'")
	}
	s.table = table

	// 'column_mapping' 是必需配置。
	mappingVal, ok := config["column_mapping"]
	if !ok {
		return fmt.Errorf("mysql sink: config is missing required key 'column_mapping'")
	}

	// YAML 解析器会将 map 解析为 map[interface{}]interface{}，这里需要进行类型转换。
	mapping, ok := mappingVal.(map[interface{}]interface{})
	if !ok {
		return fmt.Errorf("mysql sink: 'column_mapping' is not a valid map, actual type: %T", mappingVal)
	}

	if len(mapping) == 0 {
		return fmt.Errorf("mysql sink: 'column_mapping' cannot be empty")
	}

	s.columnMapping = make(map[string]string)
	for k, v := range mapping {
		kStr, okK := k.(string)
		vStr, okV := v.(string)
		if !okK || !okV {
			return fmt.Errorf("mysql sink: 'column_mapping' keys and values must all be strings")
		}
		s.columnMapping[kStr] = vStr
	}

	var err error
	s.db, err = sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("mysql sink: failed to initialize database connection pool: %w", err)
	}

	if err = s.db.Ping(); err != nil {
		return fmt.Errorf("mysql sink: failed to connect to database: %w", err)
	}

	return nil
}

// Write 将一批记录通过构建一个大的 INSERT 语句在事务中批量写入 MySQL。
//
// 设计说明:
//  1. **事务性**: 整个批次在一个数据库事务中执行，保证了操作的原子性（要么全部成功，要么全部失败）。
//  2. **批量插入**: 它动态构建一个 `INSERT INTO ... VALUES (...), (...), ...` 语句，
//     这比逐条 INSERT 效率高得多，因为它减少了网络往返和数据库开销。
//  3. **安全性**: 所有值都通过参数化查询传递，使用了 `?` 作为占位符，这可以有效防止 SQL 注入攻击。
//  4. **列映射**: `column_mapping` 配置决定了 `core.Record` 中的哪个字段写入数据库的哪个列。
//  5. **空值处理**: 如果一条记录中缺少 `column_mapping` 中指定的某个键，该字段在插入时将被视为 `NULL`。
func (s *Sink) Write(records []core.Record) error {
	if len(records) == 0 {
		return nil
	}

	if s.db == nil {
		return fmt.Errorf("mysql sink: database connection is not open")
	}

	// 启动事务
	tx, err := s.db.Begin()
	if err != nil {
		return fmt.Errorf("mysql sink: failed to begin transaction: %w", err)
	}
	// 如果出现任何错误，确保事务被回滚。
	defer tx.Rollback()

	// 根据 column_mapping 准备数据库列名和占位符顺序。
	dbColumns := make([]string, 0, len(s.columnMapping))
	placeholders := make([]string, 0, len(s.columnMapping))
	recordKeysInOrder := make([]string, 0, len(s.columnMapping))

	for recordKey, dbCol := range s.columnMapping {
		dbColumns = append(dbColumns, "`"+dbCol+"`") // 为列名加上反引号以处理保留字
		placeholders = append(placeholders, "?")
		recordKeysInOrder = append(recordKeysInOrder, recordKey)
	}

	// 构建 SQL 语句
	// 最终形式: "INSERT INTO `table` (`col1`, `col2`) VALUES (?, ?), (?, ?), ..."
	valuePlaceholderGroup := "(" + strings.Join(placeholders, ", ") + ")"
	allValuePlaceholders := strings.Repeat(valuePlaceholderGroup+",", len(records)-1) + valuePlaceholderGroup

	query := fmt.Sprintf("INSERT INTO `%s` (%s) VALUES %s", s.table, strings.Join(dbColumns, ", "), allValuePlaceholders)

	// 准备所有参数
	args := make([]interface{}, 0, len(records)*len(s.columnMapping))
	for _, record := range records {
		for _, recordKey := range recordKeysInOrder {
			val, exists := record[recordKey]
			if !exists {
				val = nil // 如果记录中缺少该键，则插入 NULL
			}
			args = append(args, val)
		}
	}

	// 执行批量插入
	_, err = tx.Exec(query, args...)
	if err != nil {
		return fmt.Errorf("mysql sink: failed to execute batch insert: %w", err)
	}

	// 如果一切顺利，提交事务。
	return tx.Commit()
}

// Close 负责关闭数据库连接池，释放所有底层连接。
func (s *Sink) Close() error {
	if s.db != nil {
		return s.db.Close()
	}
	return nil
}
